package com.unism.unisdk.core.transport.invoke;

import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.ning.http.client.AsyncHttpClient;
import com.ning.http.client.AsyncHttpClientConfig;
import com.unism.unisdk.common.enumObj.HttpMethod;
import com.unism.unisdk.common.enumObj.Protocol;
import com.unism.unisdk.core.DefaultConfig;
import com.unism.unisdk.core.transport.ApiAttr;
import com.unism.unisdk.core.transport.OpenResponse;
import com.unism.unisdk.core.transport.OpenResponseCallback;
import com.unism.unisdk.core.transport.RequestData;
import com.unism.unisdk.core.transport.action.http.HttpAction;
import com.unism.unisdk.core.transport.action.http.HttpRequestData;

public class HttpFlowControlInvoke implements RemoteInvoke {
	public static final Protocol protocol =Protocol.HTTP;
	private static final Logger mainLogger = LoggerFactory.getLogger(HttpFlowControlInvoke.class);
	private static final Logger sdkLogger = LoggerFactory.getLogger("DING_OPEN_CLIENT_SDK_LOGGER");
	private AsyncHttpClient asyncHttpClient;
	private BlockingQueue<Runnable> queue;
	private ScheduledThreadPoolExecutor requestPool;
	private ReentrantLock lock;
	private long requestTime = 0L;
	private int sendRequestPeriod;
	private long taskNum;

	public HttpFlowControlInvoke(final DefaultConfig defaultConfig, int printQueueSizePeriod) {
		this.taskNum = 0L;
		this.lock = new ReentrantLock();
		this.requestPool = new ScheduledThreadPoolExecutor(defaultConfig.getFolwControl().getThreadPoolSize());
		this.queue = this.requestPool.getQueue();
		this.sendRequestPeriod = (60000 / defaultConfig.getFolwControl().getMaxQpm());
		this.requestPool.scheduleAtFixedRate(new Runnable() {
			public void run() {
				HashMap<String,Object> base = new HashMap<String,Object>();
				base.put("max_qpm", Integer.valueOf(defaultConfig.getFolwControl().getMaxQpm()));
				base.put("send_request_period", HttpFlowControlInvoke.this.sendRequestPeriod + "ms");

				Map<String,Object> requestQueue = new HashMap<String,Object>();
				requestQueue.put("size", Integer.valueOf(HttpFlowControlInvoke.this.queue.size()));
				requestQueue.put("remainingCapacity",
						Integer.valueOf(HttpFlowControlInvoke.this.queue.remainingCapacity()));

				Map<String,Object> threadPool = new HashMap<String,Object>();
				threadPool.put("core_size", Integer.valueOf(HttpFlowControlInvoke.this.requestPool.getCorePoolSize()));
				threadPool.put("current_size", Integer.valueOf(HttpFlowControlInvoke.this.requestPool.getPoolSize()));
				threadPool.put("active_size", Integer.valueOf(HttpFlowControlInvoke.this.requestPool.getActiveCount()));
				threadPool.put("max_size", Integer.valueOf(HttpFlowControlInvoke.this.requestPool.getMaximumPoolSize()));
				threadPool.put("largest_size",
						Integer.valueOf(HttpFlowControlInvoke.this.requestPool.getLargestPoolSize()));

				Map<String,Object> task = new HashMap<String,Object>();
				task.put("all_count", Long
						.valueOf(HttpFlowControlInvoke.this.requestPool.getTaskCount() - HttpFlowControlInvoke.this.taskNum));
				task.put("completed_count", Long.valueOf(HttpFlowControlInvoke.this.requestPool.getCompletedTaskCount()
						- HttpFlowControlInvoke.this.taskNum + 1L));

				Map<String,Object> logMap = new HashMap<String,Object>();
				logMap.put("base", base);
				logMap.put("request_queue", requestQueue);
				logMap.put("thread_pool", threadPool);
				logMap.put("task", task);
				HttpFlowControlInvoke.sdkLogger.info(JSON.toJSONString(logMap));
			}
		}, 0L, printQueueSizePeriod, TimeUnit.SECONDS);

		AsyncHttpClientConfig.Builder builder = new AsyncHttpClientConfig.Builder();
		builder.setConnectTimeout(defaultConfig.getGlobalHttpConfig().getConnectTimeout());
		builder.setReadTimeout(defaultConfig.getGlobalHttpConfig().getReadTimeout());
		builder.setRequestTimeout(defaultConfig.getGlobalHttpConfig().getRequestTimeout());
		builder.setMaxRequestRetry(defaultConfig.getGlobalHttpConfig().getMaxRequestRetry());
		this.asyncHttpClient = new AsyncHttpClient(builder.build());
	}

	private long getTime() {
		try {
			this.lock.lock();
			long t = System.currentTimeMillis();
			if (this.requestTime == 0L) {
				this.requestTime = t;
			} else if (t - this.requestTime < this.sendRequestPeriod)
				this.requestTime += this.sendRequestPeriod;
			else {
				this.requestTime = t;
			}
			return this.requestTime;
		} finally {
			this.lock.unlock();
		}
	}

	public OpenResponse execute(final RequestData requestData) throws Exception {
		final HttpRequestData httpRrquest=(HttpRequestData) requestData;
		
		httpRrquest.recordStartExecuteTime();
		final HttpAsyncRequestTask requestTask = new HttpAsyncRequestTask();
		Long expectedExcuteTime = getTime();
		httpRrquest.setExpectedQuitQueueTime(expectedExcuteTime);
		this.requestPool.schedule(new Runnable() {
			public void run() {
				try {
					requestData.recordActualQuitQueueTime();
					
					ApiAttr apiAttr = httpRrquest.getApiAttr();//api接口注解信息
					String fullUrl = httpRrquest.getFullUrl();//url
					String  body = httpRrquest.getBody();//body内容,必须都是字符穿格式
					String requestId = requestData.getRequestId();
					String finalPostFileKey = httpRrquest.getFileKey();
					File finalPostFile = httpRrquest.getFile();
					HttpMethod httpMethod=apiAttr.getHttpMethod();
					HashMap<String,String> header=httpRrquest.getHeadParam();
					
					HttpAction.httpRequest(requestTask.getHttpCallBack(),
							asyncHttpClient, requestId, apiAttr.getRequestTimeout(), httpMethod,fullUrl, body,
							header, finalPostFileKey, finalPostFile);
				} catch (Exception e) {
					OpenResponse openResponse = new OpenResponse();
					openResponse.setThrowable(e);
					requestTask.getHttpCallBack().onCallBack(openResponse);
				}
			}
		}, expectedExcuteTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
		requestTask.pause();

		requestData.recordEndExecuteTime();
		return requestTask.getOpenResponse();
	}

	public void close() {
		this.requestPool.shutdown();
		this.asyncHttpClient.close();
	}

	private class HttpAsyncRequestTask {
		private CountDownLatch latch;
		private OpenResponse openResponse;
		private OpenResponseCallback httpCallBack;

		public HttpAsyncRequestTask() {
			this.latch = new CountDownLatch(1);
			this.httpCallBack = new OpenResponseCallback() {
				public void onCallBack(OpenResponse response) {
					HttpFlowControlInvoke.HttpAsyncRequestTask.this.openResponse = response;
					HttpFlowControlInvoke.HttpAsyncRequestTask.this.latch.countDown();
				}
			};
		}

		public void pause() throws InterruptedException {
			this.latch.await();
		}

		public OpenResponse getOpenResponse() {
			return this.openResponse;
		}

		public OpenResponseCallback getHttpCallBack() {
			return this.httpCallBack;
		}
	}
}